package org.finesys.common.hazelcast.core.mq;


import com.hazelcast.collection.IQueue;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.topic.ITopic;
import lombok.RequiredArgsConstructor;

import java.util.List;

@RequiredArgsConstructor
public class HazelcastMQTemplate {
    private final HazelcastInstance hazelcastInstance;
    private final List<HazelcastMessageInterceptor> interceptorList;

    /**
     * 发送分组消息
     *
     * @param message 消费者组消息
     * @param <T>     消费者组消息
     */
    public <T extends AbstractGroupMessage> void send(T message) {
        String groupName = message.getGroup();
        IQueue<T> queue = hazelcastInstance.getQueue(groupName);
        try {
            sendMessageBefore(message);
            queue.put(message);
            sendMessageAfter(message);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 发送广播消息
     *
     * @param message 广播消息
     * @param <T>     广播消息
     */
    public <T extends AbstractTopicMessage> void send(T message) {
        String topic = message.getTopic();
        ITopic<T> tiTopic = hazelcastInstance.getTopic(topic);
        tiTopic.publish(message);
    }

    private <T extends AbstractGroupMessage> void sendMessageAfter(T message) {
        for (int i = interceptorList.size() - 1; i >= 0; i--) {
            interceptorList.get(i).sendMessageAfter(message);
        }

    }

    private <T extends AbstractGroupMessage> void sendMessageBefore(T message) {
        interceptorList.forEach(interceptor -> interceptor.sendMessageBefore(message));
    }
}
